package com.example.toy.core.commit;

import com.example.toy.core.utils.SchemaHelper;
import com.example.toy.core.version.ToyVersion;
import java.io.ByteArrayOutputStream;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import lombok.Builder;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import org.apache.avro.Schema;

/**
 * CommitMsg
 *
 * @author shenbing
 * @date 2021/12/17
 */
@Getter
@Setter
@Builder(toBuilder = true)
public class CommitMsg implements Comparable<CommitMsg> {

  private String msg;
  private ToyVersion version;
  private Schema schema;
  private CommitState commitState;
  private CommitType commitType;
  private long timestamp;
  private Map<String, String> properties;

  public CommitMsg() {
    this.timestamp = System.currentTimeMillis();
    this.properties = new HashMap<>();
  }

  public CommitMsg(String msg, ToyVersion version, Schema schema,
      CommitState commitState, CommitType commitType, long timestamp,
      Map<String, String> properties) {
    this.msg = msg;
    this.version = version;
    this.schema = schema;
    this.commitState = commitState;
    this.commitType = commitType;
    this.timestamp = timestamp;
    this.properties = properties;
  }

  public void addAttribute(String name, String value) {
    this.properties.put(name, value);
  }

  public static CommitMsg deserailize(Properties properties) {
    Map<String, String> p = new HashMap<>();
    Set<String> propertyKeys = properties.stringPropertyNames();
    Arrays.asList("msg", "version", "schema", "commitType", "timestamp")
        .forEach(propertyKeys::remove);
    for (String propertyKey : propertyKeys) {
      p.put(propertyKey, properties.getProperty(propertyKey));
    }

    CommitMsg commit =
        CommitMsg.builder()
            .msg(properties.getProperty("msg"))
            .version(new ToyVersion(Long.parseLong(properties.getProperty("version"))))
            .schema(SchemaHelper.fromString(properties.getProperty("schema")))
            .commitType(CommitType.valueOf(properties.getProperty("commitType")))
            .timestamp(Long.parseLong(properties.getProperty("timestamp")))
            .properties(p)
            .build();
    return commit;
  }

  /**
   * 序列化成字节
   *
   * @return
   */
  @SneakyThrows
  public byte[] serialize() {
    Properties properties = new Properties();
    properties.put("msg", getMsg() == null ? "" : getMsg());
    properties.put("version", String.valueOf(getVersion().getTimestamp()));
    properties.put("schema", getSchema() == null ? "" : SchemaHelper.toString(getSchema()));
    properties.put("commitType", getCommitType().name());
    properties.put("timestamp", String.valueOf(getTimestamp()));
    if (this.properties != null) {
      properties.putAll(this.properties);
    }
    try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
      properties.store(byteArrayOutputStream, "serialize");
      return byteArrayOutputStream.toByteArray();
    }
  }

  @Override
  public int compareTo(CommitMsg o) {
    if (o == null) return -1;
    if (this.version == null) {
      return 1;
    }
    return this.getVersion().compareTo(o.getVersion());
  }
}
